package com.xuhf;

import cn.hutool.core.date.DateUtil;
import com.alibaba.fastjson.JSONObject;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.FailureCallback;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.SuccessCallback;

import javax.annotation.Resource;

@Component
public class MessageProducer {

    private String TOPIC_NAME = "my_test";

    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;

    @Scheduled(fixedRate = 5 * 1000)
    public void send() {
        ListenableFuture<SendResult<String, String>> listenableFuture = kafkaTemplate.send(TOPIC_NAME, "消息【" + DateUtil.now() + "】");
        //发送成功后回调
        SuccessCallback successCallback = result -> System.out.println("发送成功:" + JSONObject.toJSONString(result));
        //发送失败回调
        FailureCallback failureCallback = ex -> System.err.println("发送失败:" + JSONObject.toJSONString(ex));
        listenableFuture.addCallback(successCallback, failureCallback);
    }
}
